package org.study.mq.kafka.java;

import org.apache.kafka.clients.producer.*;
import org.apache.log4j.Logger;


import java.util.HashMap;
import java.util.Map;

public class ProducerSample2 {

    private static Logger logger = Logger.getLogger(ProducerSample2.class);


    public static void main(String[] args) {
        Map<String, Object> props = new HashMap<>();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");


        String topic = "test-topic";
        Producer<String, String> producer = new KafkaProducer<String, String>(props);
        //立即发送 只需要将消息发送到服务端 不关心消息发送的结果
        producer.send(new ProducerRecord(topic, "idea-key2", "java-message 1"));

        //同步发送 获取方法返回的Future对象，根据结果查看send方法是否调用成功
        try{
            producer.send(new ProducerRecord(topic, "idea-key2", "java-message 2")).get();
        }catch (Exception e){
            logger.error("消息发送失败",e);
        }

        //异步发送
        producer.send(new ProducerRecord(topic, "idea-key2", "java-message 3"), new Callback() {
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                if (e != null){
                    logger.error("消息发送失败",e);
                }
            }
        });

        producer.close();
    }



}
